package com.gengzp.mq.server.facade.impl;

import com.fasterxml.jackson.core.type.TypeReference;
import com.gengzp.mq.common.model.dto.GengzpMqConsumerDto;
import com.gengzp.mq.common.model.entity.GengzpMqTopic;
import com.gengzp.mq.common.utils.GengzpMqAssertUtils;
import com.gengzp.mq.common.utils.GengzpMqJsonUtils;
import com.gengzp.mq.common.utils.GengzpMqRedisUtils;
import com.gengzp.mq.common.utils.GengzpMqSnowflakeIdUtils;
import com.gengzp.mq.server.facade.IGengzpMqTopicFacadeService;
import com.gengzp.mq.server.mapper.GengzpMqTopicMapper;
import com.google.common.collect.Lists;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.Instant;
import java.util.Date;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.TimeUnit;

import static com.gengzp.mq.common.constants.GengzpMqCacheConstants.TOPIC_CONSUMER_SERVICE_NAME_KEY_PREFIX;
import static com.gengzp.mq.common.constants.GengzpMqCacheConstants.TOPIC_INFO_KEY_PREFIX;

/**
 * @ClassName GengzpMqTopicFacadeServiceImpl
 * @Description 消息队列主题管理门面层实现类
 * @Author gengzp
 * @Date 2025/7/10 13:57
 */
@Service
public class GengzpMqTopicFacadeServiceImpl implements IGengzpMqTopicFacadeService {

    @Autowired
    private GengzpMqTopicMapper gengzpMqTopicMapper;

    @Override
    public GengzpMqTopic searchTopicByName(String topicName) {
        GengzpMqAssertUtils.isNotBlank(topicName, "topicName不能为空");
        return gengzpMqTopicMapper.selectTopicByName(topicName);
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public void addTopic(String topicName, String description) {
        GengzpMqTopic mqTopic = new GengzpMqTopic();
        mqTopic.setId(GengzpMqSnowflakeIdUtils.nextId());
        mqTopic.setTopic(topicName);
        mqTopic.setDescription(description);

        // 赋值时间字段
        Date date = new Date();
        mqTopic.setCreatedTime(date);
        mqTopic.setLastModifiedTime(date);

        // 缓存到redis
        GengzpMqRedisUtils.set(TOPIC_INFO_KEY_PREFIX + mqTopic.getTopic(),
                GengzpMqJsonUtils.toJson(mqTopic), 1, TimeUnit.DAYS);

        gengzpMqTopicMapper.insertTopic(mqTopic);
    }

    @Override
    public void saveTopicRelationConsumer(String topic, String consumerGroup, String consumerServiceName, String consumerServiceIp) {
        // 当前系统时间
        Date currentDate = new Date();

        GengzpMqConsumerDto mqConsumerDto = new GengzpMqConsumerDto();
        mqConsumerDto.setTopic(topic);
        mqConsumerDto.setConsumerServiceName(consumerServiceName);
        mqConsumerDto.setConsumerGroup(consumerGroup);
        mqConsumerDto.setConsumerServiceIp(consumerServiceIp);
        mqConsumerDto.setSubscribeStartTime(currentDate);
        mqConsumerDto.setSubscribeEndTime(
                Date.from(Instant.now().plusSeconds(35)));

        // 从redis查询出当前topic的所有消费者服务信息
        Object topicRelationConsumerObj = GengzpMqRedisUtils.get(TOPIC_CONSUMER_SERVICE_NAME_KEY_PREFIX + topic);

        // 没有人订阅此topic, 直接保存
        if (Objects.isNull(topicRelationConsumerObj)) {
            GengzpMqRedisUtils.set(TOPIC_CONSUMER_SERVICE_NAME_KEY_PREFIX + topic,
                    GengzpMqJsonUtils.toJson(Lists.newArrayList(mqConsumerDto)), 30, TimeUnit.SECONDS);
        } else {
            // 有人订阅, 则先移除掉过时的队列中订阅者, 再将当前消费者加入订阅者队列中
            List<GengzpMqConsumerDto> gengzpMqConsumerDtos = GengzpMqJsonUtils.parseObjectWithGeneric(topicRelationConsumerObj.toString(), new TypeReference<>() {
            });
            gengzpMqConsumerDtos = gengzpMqConsumerDtos.stream()
                    .filter(v -> v.getSubscribeEndTime().before(currentDate) ||
                            (StringUtils.equals(v.getConsumerServiceName(), consumerServiceName) &&
                                    StringUtils.equals(v.getConsumerGroup(), consumerGroup) &&
                                    StringUtils.equals(v.getConsumerServiceIp(), consumerServiceIp))).toList();
            List<GengzpMqConsumerDto> finalGengzpMqConsumerDtos = Lists.newArrayList(gengzpMqConsumerDtos);
            finalGengzpMqConsumerDtos.add(mqConsumerDto);
            GengzpMqRedisUtils.set(TOPIC_CONSUMER_SERVICE_NAME_KEY_PREFIX + topic,
                    GengzpMqJsonUtils.toJson(finalGengzpMqConsumerDtos), 30, TimeUnit.SECONDS);
        }

    }

    @Override
    public void removeTopicRelationConsumer(String topic, String consumerServiceName, String consumerServiceIp) {
        // 从redis查询出当前topic的所有消费者服务信息
        Object topicRelationConsumerObj = GengzpMqRedisUtils.get(TOPIC_CONSUMER_SERVICE_NAME_KEY_PREFIX + topic);
        if (Objects.isNull(topicRelationConsumerObj)) {
            return;
        }

        // 系统当前时间
        Date currentDate = new Date();
        List<GengzpMqConsumerDto> gengzpMqConsumerDtos = GengzpMqJsonUtils.parseObjectWithGeneric(topicRelationConsumerObj.toString(), new TypeReference<>() {
        });
        gengzpMqConsumerDtos = gengzpMqConsumerDtos.stream()
                .filter(v -> (StringUtils.equals(v.getConsumerServiceName(), consumerServiceName) &&
                        StringUtils.equals(v.getConsumerServiceIp(), consumerServiceIp)) ||
                        v.getSubscribeEndTime().before(currentDate)).toList();
        if (CollectionUtils.isEmpty(gengzpMqConsumerDtos)) {
            GengzpMqRedisUtils.delete(TOPIC_CONSUMER_SERVICE_NAME_KEY_PREFIX + topic);
        } else {
            GengzpMqRedisUtils.set(TOPIC_CONSUMER_SERVICE_NAME_KEY_PREFIX + topic,
                    GengzpMqJsonUtils.toJson(gengzpMqConsumerDtos), 30, TimeUnit.SECONDS);
        }
    }

}
